package org.andy.zookeeper;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;

import org.andy.util.ObjectUtil;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * ZooKeeper Java Api 使用样例<br>
 * version 3.4.5
 *
 * @author andy
 * @version V1.0
 * @date 18/2/14 下午4:49
 */
public class JavaApiSimple implements Watcher {

    private static final Logger LOG = LoggerFactory.getLogger(JavaApiSimple.class);

    private static final String ADDRESS = "localhost:2181";
    private static final String ZK_PATH = "/andy";
    private static final int TIMEOUT = 10000;

    private ZooKeeper zk = null;

    private CountDownLatch latch = new CountDownLatch(1);

    /**
     * 创建连接
     *
     * @param address zk服务地址
     * @param timeout 连接超时时间
     */
    public void createConnection(String address, int timeout) {
        closeConnection();
        try {
            zk = new ZooKeeper(address, timeout, this);
            latch.await();

        } catch (IOException | InterruptedException e) {
            LOG.error("create connection failure {}", e.getMessage());
        }

    }

    /**
     * 创建节点
     *
     * @param path 节点路径
     * @param data 初始数据内容
     * @return boolean
     */
    public boolean createNode(String path, String data) {
        try {
            String result = zk.create(path, data.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

            LOG.info("节点创建成功, Path:{} ,content:{}", result, data);

        } catch (KeeperException | InterruptedException e) {
            LOG.error("create node failure {}", e.getMessage());
        }
        return true;
    }

    public String readData(String path) {
        String data = "";
        try {
            data = new String(zk.getData(path, false, null));

            LOG.info("数据读取成功， path : {}", path);
        } catch (KeeperException | InterruptedException e) {
            LOG.error("getData failure {}", e.getMessage());
        }
        return data;
    }

    public boolean writeData(String path, String data) {
        try {
            Stat stat = zk.setData(path, data.getBytes(), -1);

            LOG.info("更新数据成功，path：{} , stat: {}", path, stat);

        } catch (KeeperException | InterruptedException e) {
            LOG.error("setData failure {}", e.getMessage());
        }
        return true;
    }

    public void deleteNode(String path) {
        try {
            zk.delete(path, -1);
        } catch (InterruptedException | KeeperException e) {
            LOG.error("deleteNode failure {}", e.getMessage());
        }
    }

    public void closeConnection() {
        if (ObjectUtil.notNull(zk)) {
            try {
                zk.close();
            } catch (InterruptedException e) {
                LOG.error("close zk connection failure {}", e.getMessage());
            }
        }
    }

    @Override
    public void process(WatchedEvent event) {
        LOG.info("收到事件通知：{} \n", event.getState());

        if (KeeperState.SyncConnected == event.getState()) {
            latch.countDown();
        }
    }

    public static void main(String[] args) {
        JavaApiSimple javaApiSimple = new JavaApiSimple();
        javaApiSimple.createConnection(ADDRESS, TIMEOUT);
        boolean isCreate = javaApiSimple.createNode(ZK_PATH, "初始化数据");
        if (isCreate) {
            LOG.info("数据内容：{}", javaApiSimple.readData(ZK_PATH));
            LOG.info("更新：{}", javaApiSimple.writeData(ZK_PATH, "更新后"));
            LOG.info("更新后内容：{}", javaApiSimple.readData(ZK_PATH));
        }
        javaApiSimple.closeConnection();
    }
}
